package myEddy.operator;

import java.util.ArrayList;
import java.util.BitSet;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import myEddy.tuple.Tuple;

/**
 * Join operator.
 * It works as a symmetric hash join
 * @author Nathan Ramoly
 *
 */
public class Join extends Operator 
{
        /** Attributes **/
        
        /**
         * Column the join is made on
         */
        String join_column_1;
        String join_column_2;
        
        /**
         * Hashtables (use hash code to deal with)
         */
        Map<Integer, ArrayList<Tuple>> hashtable1;
        Map<Integer, ArrayList<Tuple>> hashtable2;
        
        /** methods **/
        
        /**
         * Constructor
         */
        public Join(String table1, String table2, String column1, String column2)
        {
                //Adding the tables of working
                tables = new ArrayList<String>();
                tables.add(table1);
                tables.add(table2);
                
                join_column_1 = column1;
                join_column_2 = column2;
                
                hashtable1 = new HashMap<Integer, ArrayList<Tuple>>(); 
                hashtable2 = new HashMap<Integer, ArrayList<Tuple>>(); 
                
                //Creating the input list
                input = new LinkedBlockingQueue<Tuple>();
                
                running = false;
        }
        
        /**
         * Main loop of the operator.
         * Read the inputs using the other methods
         */
        @SuppressWarnings("static-access")
        @Override
        public void run() 
        {
                running = true;
                long start_time = 0;
                long cur_time = 0;
                
                //Check if eddy is set
                if(parent == null)
                { } //Handle error
                
                while(running)
                {
                        //If queue not empty, getting the last one to operate it
                        if(input.size() > 0)
                        {
                                //Compute cost
                                start_time = System.currentTimeMillis();
                                
                                //Operate
                                Tuple to_operate = input.poll();
                                
                                //Operation...
                                ArrayList<Tuple> res = Operate(to_operate);
                                
                                //Sending it to eddy !
                                if(res != null)
                                { sendResultTuple(res); }
                                
                                //Updating cost, to improve
                                cur_time = System.currentTimeMillis();                          
                                perf_ind = (int) (cur_time - start_time);
                                
                        }
                        //If no tuple available, wait for some.
                        else
                        {
                                try {
                                        Thread.currentThread().sleep(50);
                                } catch (InterruptedException e) {
                                        e.printStackTrace();
                                }
                        }
                }               
                //Thread end
        }

        
        /**
         * Do the symmetric join
         * @param T New tuple to deal with
         * @return All matching tuple 
         */
        @Override
        public ArrayList<Tuple> Operate(Tuple T) 
        {
                
                Map<Integer, ArrayList<Tuple>> toPutIn = null;
                Map<Integer, ArrayList<Tuple>> toHash = null;
                String colToPut = new String();
                String colToHash = new String();
                
                //Detect which table it goes in
                if( T.getTables().contains( tables.get(0) ) ) //In table 1
                {
                        toPutIn = hashtable1;
                        toHash = hashtable2;
                        colToPut = join_column_1;
                        colToHash = join_column_2;
                }
                else if ( T.getTables().contains( tables.get(1)) ) // in table 2
                {
                        toPutIn = hashtable2;
                        toHash = hashtable1;
                        colToPut = join_column_2;
                        colToHash = join_column_1;
                }
                else
                {
                        //bug                   
                }
                
                //Hash
                int hash_val = T.getElement( colToPut ).hashCode();
                
                //Put it in the hash table
                if( toPutIn.get(hash_val) == null)
                {
                        toPutIn.put(hash_val, new ArrayList<Tuple>() );
                }
                
                toPutIn.get(hash_val).add(T);
                
                
                //Hash the other one
                ArrayList<Tuple> cur_hastable = toHash.get(hash_val);
                
                ArrayList<Tuple> ret = new ArrayList<Tuple>();
                
                //Going through the hastable and hash it        
                if( cur_hastable != null )
                {
                        for(int i=0; i<cur_hastable.size(); i++)
                        {
                                //hash
                                Tuple hashed_tuple = cur_hastable.get(i);
                                
                                //If it is the same value, out
                                if( hashed_tuple.getElement(colToHash).equals( T.getElement(colToPut) ) )
                                {
                                        //From orig tuple,
                                        Tuple new_tuple = new Tuple(T);
                                        
                                        //Add the elements
                                        Set<String> columns = hashed_tuple.getColumns();
                                        
                                        //Joining
                                        for( String column : columns )
                                        {
                                                new_tuple.addElement(column, hashed_tuple.getElement(column) );
                                        }
                                        
                                        //update bitmap
                                        BitSet done_BM = new_tuple.getDoneBitmap();
                                        BitSet done_BM_2 = hashed_tuple.getDoneBitmap();
                                        done_BM.or(done_BM_2);
                                        new_tuple.setDoneBitmap(done_BM);
                                        
                                        //update tables of the tuples (add the tables of the hashed tuple)
                                        for(int j=0; j<hashed_tuple.getTables().size(); j++)
                                        {
                                                new_tuple.addTable( hashed_tuple.getTables().get(j) );
                                        }
                                        
                                        ret.add(new_tuple);
                                }
                        }
                }
                
                return ret;
        }

        /**
         * Return the performance of the current operator
         * Size of the index for the time being
         * @return perf_ind
         * @see perf_ind
         */
        @Override
        public int getPerfInd() {
                //return perf_ind;
        		return (hashtable1.size() + hashtable2.size())/2;
        }

        /**
         * Send the result to Eddy and updates the bitmaps according.
         * @param results ArrayList of tuple to send to Eddy
         */
        @Override
        void sendResultTuple( ArrayList<Tuple> results ) {
                
                for(int i=0; i<results.size(); i++)
                {
                        Tuple T = results.get(i);
                        
                        //Updating bitmaps
                        BitSet TDoneBM = T.getDoneBitmap();
                        BitSet TToDoBM = T.getToDoBitmap();
                        
                        TDoneBM.set(id, true);
                        TToDoBM.set(id, false);
                        
                        T.setDoneBitmap(TDoneBM);
                        T.setToDoBitmap(TToDoBM);
                        
                        //Send to eddy
                        parent.getBackTuple(T, id);
                }               
        }

}
